package org.rency.mq.activemq.listener.factory.concurrent;

import org.rency.mq.activemq.listener.factory.AbstractMessageListener;
import org.rency.mq.activemq.listener.factory.MessageListenerContainer;

import javax.jms.MessageListener;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConcurrentMessageListener extends AbstractMessageListener implements MessageListener, MessageHandleCallback {
	private MessageListenerContainer messageListenerContainer;

	private ExecutorService executorService;
	private ConcurrentMap<String, MessageHandleTask> runningTasks = new ConcurrentHashMap<String, MessageHandleTask>();
	private int maxThreadsHold;
	private int coreThreadsHold;
	private boolean toBeStared = true;
	private boolean toBeStopped = false;

	@Override
	public void handleRequest(Object request) {
		MessageHandleTask task = new MessageHandleTask(getMessageHandler(),
				request, this);

		addTask(task);
		getExecutorService().execute(task);

		// 如果总线程数超过maxThreadsHold,则停止接收新的消息.
		checkRunningThreads();
	}

	public void handleRequestFinished(String taskId, boolean succ) {
		runningTasks.remove(taskId);
		
		//如果总线程数小于maxThreadsHold,并且小于coreThreadsHold,则重新接收新的消息.
		checkRunningThreads();
	}

	private void addTask(MessageHandleTask task) {
		this.runningTasks.put(task.getTaskId(), task);
	}

	private synchronized void checkRunningThreads() {
		int size = runningTasks.size();
		if (size >= maxThreadsHold && !toBeStopped) {
			fireStopListeningEvent();
			toBeStopped = true;
			toBeStared = false;
		} else if (size < coreThreadsHold && !toBeStared) {
			fireStartListeningEvent();
			toBeStared = true;
			toBeStopped = false;
		}
	}

	private void fireStopListeningEvent() {
		messageListenerContainer.stopContainer();
	}

	private void fireStartListeningEvent() {
		messageListenerContainer.startContainer();
	}

	public MessageListenerContainer getMessageListenerContainer() {
		return messageListenerContainer;
	}

	public void setMessageListenerContainer(
			MessageListenerContainer messageListenerContainer) {
		this.messageListenerContainer = messageListenerContainer;
	}

	public ExecutorService getExecutorService() {
		if(executorService == null){
			executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2 + 1);
		}
	
		return executorService;
	}

	public void setExecutorService(ExecutorService executorService) {
		this.executorService = executorService;
	}

	public int getMaxThreadsHold() {
		return maxThreadsHold;
	}

	public void setMaxThreadsHold(int maxThreadsHold) {
		this.maxThreadsHold = maxThreadsHold;
	}

	public int getCoreThreadsHold() {
		return coreThreadsHold;
	}

	public void setCoreThreadsHold(int coreThreadsHold) {
		this.coreThreadsHold = coreThreadsHold;
	}

}
